package com.kwin.rpcclient.holder;

import com.kwin.rpcclient.rpc.NettyClient;
import com.kwin.rpccommon.dto.Request;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author : wangkai
 * @version V1.0
 * @Package com.kwin.rpcclient.holder
 * @Description:
 * @date 2021年08月03日 10:15
 **/
@Slf4j
public class ConnManage {
    
    private static NettyClient nettyClient = new NettyClient();
    
    private static Map<String, AtomicInteger> roundRobin = new ConcurrentHashMap<>();
    private static Map<String, CopyOnWriteArrayList<Channel>> servChannel = new ConcurrentHashMap<>();
    private static Map<SocketAddress, Channel> channelNodes = new ConcurrentHashMap<>();
    
    /**
     * 针对可用连接进行线性轮询负载
     * @return
     */
    public static Channel chooseChannel(String name) {
        CopyOnWriteArrayList<Channel> channels = servChannel.get(name);
        AtomicInteger robin = roundRobin.get(name);
        if (channels.size() > 0) {
            int size = channels.size();
            int index = (robin.getAndAdd(1) + size) % size;
            
            log.info("size is {}", size);
            return channels.get(index);
        }else{
            return null;
        }
    }
    
    public static void updateConnServer(String name, List<String> addrs) {
        if (addrs.size() == 0 || addrs == null) {
            log.error("没有可用的服务器节点, 全部服务节点已关闭!");
            CopyOnWriteArrayList<Channel> channels = servChannel.get(name);
            if(channels == null) {
                return;
            }
            for (Channel channel : channels) {
                SocketAddress remotePeer = channel.remoteAddress();
                Channel handlerNode = channelNodes.get(remotePeer);
                handlerNode.close();
            }
            channels.clear();
            channelNodes.clear();
            return;
        }
    
        HashSet<SocketAddress> serverNodeSet = new HashSet<>();
        for (int i = 0; i < addrs.size(); ++i) {
            String[] array = addrs.get(i).split(":");
            if (array.length == 2) {
                String host = array[0];
                int port = Integer.parseInt(array[1]);
                SocketAddress remotePeer = new InetSocketAddress(host, port);
                serverNodeSet.add(remotePeer);
            }
        }
    
        // 维护新连接节点
        for (final SocketAddress serverNodeAddress : serverNodeSet) {
            Channel channel = channelNodes.get(serverNodeAddress);
            if(channel != null && !channel.isOpen()) {
                CopyOnWriteArrayList<Channel> channels = servChannel.get(name);
                channels.remove(channel);
            }
            if (channel != null && channel.isOpen()){
                log.info("当前服务节点已存在,无需重新连接.{}", serverNodeAddress);
            }else{
//                if(!RpcClientHolder.lazy) {
                   connectServerNode(name, serverNodeAddress);
//                }
            }
        }
    
        CopyOnWriteArrayList<Channel> channels = servChannel.get(name);
    
        log.error("CopyOnWriteArrayList is {}", channels.size());
        // 移除失效节点
        for (int i = 0; i < channels.size(); ++i) {
            Channel channel = channels.get(i);
            SocketAddress remotePeer = channel.remoteAddress();
            log.error("SocketAddress is {}", remotePeer);
            if (!serverNodeSet.contains(remotePeer)) {
                log.info("删除失效服务节点 {}", remotePeer);
                Channel channelNode = channelNodes.get(remotePeer);
                if (channelNode != null) {
                    channelNode.close();
                }
                channels.remove(channel);
                channelNodes.remove(remotePeer);
            }
        }
        
    }
    
    private static void connectServerNode(String name, SocketAddress address){
        
        try {
            Channel channel = nettyClient.doConnect(address);
            addChannel(name, channel, address);
        } catch (InterruptedException e) {
            e.printStackTrace();
            log.info("未能成功连接到服务器:{}", address);
        }
    }
    
    private static void addChannel(String name, Channel channel, SocketAddress address) {
        log.info("加入Channel到连接管理器.{}", address);
        CopyOnWriteArrayList<Channel> channels = servChannel.get(name);
        if(channels == null) {
            channels = new CopyOnWriteArrayList<>();
            servChannel.put(name, channels);
            AtomicInteger au = new AtomicInteger(0);
            roundRobin.put(name, au);
            
        }
        channels.add(channel);
        channelNodes.put(address, channel);
    }
    
    public static Object send(String name, Request req) throws InterruptedException {
        Channel channel = chooseChannel(name);
        return  nettyClient.send(channel, req);
    }
}
